﻿using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace WalkthroughTDF
{
    /// <summary>
    /// due to AsyncCTP is still in development, the APIs are still not fixed
    /// so I don't use "async/await" here, but have to develop my own pattern to call "ReceiveAsync"
    /// just like socket's "Async-IO" mode
    /// </summary>
    sealed class TransBlockAsyncRcver
    {
        private TransformBlock<int, string> m_tb;

        public TransBlockAsyncRcver(int sleepTime, int total)
        {
            m_tb = new TransformBlock<int, string>(num => TestTransformBlock.Map(num, sleepTime));
            m_tb.Insert(total);
        }

        public void BeginReceive()
        {
            m_tb.ReceiveAsync().ContinueWith(this.EndReceiveCallback);
        }

        private void EndReceiveCallback(Task<string> compTask)
        {
            Console.WriteLine("receive in thread<{0}>: {1}", Thread.CurrentThread.ManagedThreadId, compTask.Result);
            BeginReceive();
        }
    }

    static class TestTransformBlock
    {
        public static void Insert(this TransformBlock<int, string> tb, int total)
        {
            for (int index = 0; index < total; ++index)
                tb.Post(index);
        }

        public static string Map(int num, int sleepTime)
        {
            Thread.Sleep(sleepTime);
            return string.Format("data{0}", num);
        }

        /// <summary>
        /// by default (where MaxParallelDegree=1), data is transformed asynchronously and sequentially
        /// and buffered to be retrieved
        /// </summary>
        private static void TestSyncReceive()
        {
            var tb = new TransformBlock<int, string>(num => Map(num, 500));
            int total = 10;
            Insert(tb, total);

            for (int index = 0; index < total; ++index)
            {
                string returned = tb.Receive();// chekanote: block and receive one by one
                Console.WriteLine("current returned='{0}',input count={1}", returned, tb.InputCount);
            }
        }

        private static void TestAsyncReceive()
        {
            TransBlockAsyncRcver receiver = new TransBlockAsyncRcver(500, 10);
            receiver.BeginReceive();

            Console.WriteLine("Press any key to exit, ......");
            Console.ReadLine();
        }

        public static void TestMain()
        {
            // TestSyncReceive();
            TestAsyncReceive();
        }
    }
}
